import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.spark.sql.SparkSession;
import com.meritdata.tempo.ai.module.run.utils.JSONUtils;

import com.meritdata.tempo.ai.module.run.mine.operator.datameta.SyncData;
import com.meritdata.tempo.ai.module.run.mine.server.sync.Client;
import com.meritdata.tempo.ai.module.run.mine.server.thrift.ExecuteResultBean;

/**
 * @author meritdata
 * 同步服务本地化运行调用示例
 * 服务包结构：
 * --<服务ID>
 * 		--models
 * 		--flow.xml
 * 		--LocalClientTest.java 调用示例
 *		--offline_licence.dat  校验文件
 * 
 * 将下载的服务包解压到存放服务文件的地方（即初始化的根目录文件路径）
 * 然后设置示例代码中的资源路径为解压的路径
 * 校验文件名称不能改变，否则会报错
 */
public class LocalClientTest {

	public static void main(String[] args) {
		// Auto-generated method stub
		// 1.设置服务资源的根目录
		Client client = new Client("E:\\user\\localModel");
		// 2.初始化环境SparkSession的相关配置，花费时间较长，初始化一次即可
		SparkSession spark = client.init();
		List<Map<String, Object>> rows = new ArrayList<>();
		// 3.添加每行数据，key为字段名称，value为字段值
		java.util.Map<String, Object> row = new java.util.HashMap<>();
		<#list columns as col>
		<#if col.type == 'DOUBLE'>
		row.put("${col.name}", 1);
		<#elseif col.type == 'DECIMAL'>
		row.put("${col.name}", 1);
		<#elseif col.type == 'LONG'>
		row.put("${col.name}", 1);
		<#elseif col.type == 'INT'>
		row.put("${col.name}", 1);
		<#elseif col.type == 'FLOAT'>
		row.put("${col.name}", 1);
		<#elseif col.type == 'TIMESTAMP'>
		// 设置日期时间类型为yyyy-MM-dd格式的字符类型
		row.put("${col.name}", "2018-01-01");
		<#elseif col.type == 'DATETIME'>
		// 设置日期时间类型为yyyy-MM-dd格式的字符类型
		row.put("${col.name}", "2018-01-01");
		<#else>
		row.put("${col.name}", "label");	
		</#if>
		</#list>
		rows.add(row);
        <#if flowParams?? &&(flowParams?size>0) >
        //构建流程参数信息
        Map<String, Object> flowParams = new java.util.HashMap<>();
        <#list flowParams as param>
        flowParams.put("${param.paramName}", "${param.paramValue}");
        </#list>
		// 4.调用对应的同步服务(${name})
		// 注:公共包和离线服务包需版本统一，版本不同，可能出现未知错误
		ExecuteResultBean<SyncData> r = client.run1("${service.id}", rows,flowParams);
		<#else>
		// 4.调用对应的同步服务(${name})
		ExecuteResultBean<SyncData> r = client.run("${service.id}", rows);
        </#if>
		// 5.查看结果，FINISHED表示调用成功，FAILED表示失败
		/**
		 * String status; 状态
 		 * String error; 错误信息
 		 * List<SyncData> result; 结果集
		 */
		System.out.println(r.getStatus());
		if("FINISHED".equals(r.getStatus())) {
			// 获取同步输出节点信息
			SyncData data = r.getResult().get(0);
			System.out.println("同步输出节点: " + data.getName());
			System.out.println("同步输出列：" + JSONUtils.toJsonStr(data.getColumns()));
			List<Map<String, Object>> outputRows = data.getRows();
			// 遍历输出每行数据
			for (Map<String, Object> outputRow : outputRows) {
				for(Map.Entry<String, Object> entryValue : outputRow.entrySet()) {
					System.out.print(entryValue.getKey() + ":" + entryValue.getValue() + " ");
				}
				System.out.println();
			}
		}
		// 停止spark服务，如果是要重复调用，不需要停止，否则每次调用初始化spark，会花费很长时间
		// spark.stop();
	}

}
